package com.smsc.headend.connector.manager;

import cn.hutool.core.date.SystemClock;
import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;
import com.smsc.headend.common.enums.DeviceType;
import com.smsc.headend.common.thread.UDISThreadFactory;
import com.smsc.headend.common.utils.ProtoBufferUtil;
import com.smsc.headend.connector.bo.AssociationState;
import com.smsc.headend.connector.bo.ConnectTaskResultDto;
import com.smsc.headend.connector.builder.LoggerBuilder;
import com.smsc.headend.connector.netty.NettyConfigHelper;
import com.smsc.headend.connector.service.DeviceConnectService;
import com.smsc.headend.connector.service.FeignAssetManagementService;
import com.smsc.headend.connector.service.TerminalChannelService;
import com.smsc.headend.connector.service.asset.AssetService;
import com.smsc.headend.connector.service.impl.TerminalChannelServiceImpl;
import com.smsc.headend.connector.utils.LogFileNameUtil;
import com.smsc.headend.module.asset.dmo.MeterConfigDMO;
import com.smsc.headend.module.asset.dto.KeySyncDTO;
import com.smsc.headend.module.asset.entity.ComEntity;
import com.smsc.headend.module.asset.entity.Meter;
import com.smsc.headend.module.data.dto.ClearLocalCacheDto;
import com.smsc.headend.module.protocol.cosem.util.ByteUtils;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.ConnectorTask;
import com.smsc.headend.module.task.dto.ConnectorTaskResult;
import com.smsc.headend.module.task.proto.ConnectorTaskProto;
import gurux.dlms.secure.GXDLMSSecureClient;
import io.micrometer.core.instrument.Counter;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;

@Service
@Slf4j
public class KafkaMessageManager {
    public static final String FALSE = "false";

    public static final String BATCH_FACTORY = "batchFactory";

    @Autowired
    @Qualifier("connectorTaskPool")
    ThreadPoolExecutor threadPoolExecutor;

    @Value("${con.task.device.concurrency:100}")
    Integer conNumber;

    @Autowired
    TerminalChannelService terminalChannelService;

    @Autowired
    FeignAssetManagementService feignAssetManagementService;

    @Autowired
    AssociationManager associationManager;

    @Autowired
    LogFileNameUtil logFileNameUtil;

    @Autowired
    AssetService assetService;

    @Autowired
    KafkaListenerEndpointRegistry registry;

    @Autowired
    DeviceConnectService deviceConnectService;

    @Autowired
    TaskTicketManager taskTicketManager;

    @Autowired
    @Qualifier("taskAllInboundCounter")
    Counter taskAllInCounter;

    ScheduledExecutorService registerService = Executors.newScheduledThreadPool(1, new UDISThreadFactory("register"));

    @PostConstruct
    public void initConsumerCheck() {
        //每500ms 检查任务队列
        registerService.scheduleAtFixedRate(this::checkDeviceQueue, 60 * DateUtils.MILLIS_PER_SECOND, 500, TimeUnit.MILLISECONDS);
    }

    private void checkDeviceQueue() {
        try {
            if (taskTicketManager.isTaskQueueEmpty()) {
                return;
            }
            Integer size = taskTicketManager.getTaskQueueSize();
            if (taskTicketManager.getTicketPoolSize() == 0) {
                log.info("TicketSoldOutCurrentTaskQueueSize:{}", size);
                return;
            }
            do {
                if (size % 100 == 0) {
                    log.info("LeftTaskQueueSize:{}", size);
                }
                ConnectorTask connectorTask = taskTicketManager.pollTaskQueue();
                ComEntity comEntity = terminalChannelService.getTerminalById(connectorTask.getTerminalId());
                if (comEntity != null) {
                    Integer ticketOnce = taskTicketManager.getTicketOnce(connectorTask);
                    if (ticketOnce == null) {
                        //无法获取任务凭证，等待执行中的ticket释放
                        log.info("failGetTicket,waitForRelease,TaskQueueSize:{}", size);
                        return;
                    }
                    log.info("pollConnectTask:{}", connectorTask.getTaskNo());
                    ConnectorTaskThread connectorTaskThread = new ConnectorTaskThread(connectorTask, comEntity, ticketOnce);
                    SpringUtil.getConfigurableBeanFactory().autowireBean(connectorTaskThread);
                    threadPoolExecutor.submit(connectorTaskThread);
                }
            }
            while (!taskTicketManager.isTaskQueueEmpty());
        } catch (Exception e) {
            log.error("checkTaskConsumerStatusFailed", e);
        }
    }

    @KafkaListener(topics = TaskKafkaTopic.CLOSE_ANONY_ASSET_CHANNEL, containerFactory = BATCH_FACTORY, autoStartup = FALSE)
    public void closeAnoyChannel(List<ConsumerRecord> dataList, Acknowledgment acknowledgment) {
        for (ConsumerRecord data : dataList) {
            String payload = new String((byte[]) data.value(), StandardCharsets.UTF_8);
            if (StringUtils.isNotEmpty(payload)) {

                if (payload.startsWith(TerminalChannelServiceImpl.LOGIN)) {
                    log.debug("disconnect: {}", payload);
                    String[] split = payload.split(":");
                    String moduleSn = split[1];
                    String loginCode = split[2];
                    terminalChannelService.disconnectDeprecatedChannel(moduleSn, loginCode);
                } else {
                    log.debug("closeAnoyChannel,moduleSn:{}", payload);
                    terminalChannelService.closeAnonyChannel(payload);
                }
            }
        }
        acknowledgment.acknowledge();
    }

    @KafkaListener(topics = TaskKafkaTopic.CLEAR_DEVICE_LOCAL_CACHE, autoStartup = FALSE, containerFactory = BATCH_FACTORY)
    public void clearDeviceLocalCache(List<ConsumerRecord> dataList, Acknowledgment acknowledgment) {
        log.debug("receive clear device local cache");
        for (ConsumerRecord data : dataList) {
            String payload = new String((byte[]) data.value(), StandardCharsets.UTF_8);
            ClearLocalCacheDto dto = JSONUtil.toBean(payload, ClearLocalCacheDto.class);
            if (null != dto.getMeterId()) {
                log.debug("clear local cache for meter ,meterId:{}", dto.getMeterId());
                assetService.clearMeterCacheById(dto.getMeterId());
            }
            if (null != dto.getComId()) {
                log.debug("clear local cache for comEntity ,comId:{}", dto.getComId());
                assetService.clearComEntityCacheById(dto);
            }
            if (null != dto.getDcuId()) {
                log.debug("clear local cache for dcu ,dcuId:{}", dto.getDcuId());
                assetService.clearDcuCacheById(dto.getDcuId());
            }
            acknowledgment.acknowledge();
        }
    }

    @KafkaListener(topics = TaskKafkaTopic.KMS_KEY_SYNC, containerFactory = BATCH_FACTORY, autoStartup = FALSE)
    public void updateAssociationKey(List<ConsumerRecord> dataList, Acknowledgment acknowledgment) {
        for (ConsumerRecord data : dataList) {
            String jsonStr = new String((byte[]) data.value(), StandardCharsets.UTF_8);
            KeySyncDTO keySyncDTO = JSONUtil.toBean(jsonStr, KeySyncDTO.class);
            log.debug("updateAssociationKey jsonStr={},keySyncDTO={}", jsonStr, JSONUtil.toJsonStr(keySyncDTO));
            String deviceAddress;
            ComEntity com;
            if (Objects.equals(DeviceType.METER.getId(), keySyncDTO.getDeviceType())) {
                Meter meter = feignAssetManagementService.getMeterById(keySyncDTO.getDeviceId());
                if (meter == null) {
                    log.error("the meter not exists, meter id : {}", keySyncDTO.getDeviceId());
                    return;
                }
                deviceAddress = meter.getMfgSn();
                MeterConfigDMO meterConfig = feignAssetManagementService.getMeterConfigById(meter.getMeterId());
                com = feignAssetManagementService.getComById(meterConfig.getComId());
                if (Objects.equals(com.getComDeviceType(), DeviceType.GPRS.getId())) {
                    deviceAddress = "-1";
                }
            } else {
                com = feignAssetManagementService.getComByDeviceId(keySyncDTO.getDeviceId());
                deviceAddress = "-1";
            }
            final Logger terminalLogger = LoggerBuilder.getLogger(logFileNameUtil.getLogFileName(com), this.getClass());
            if (com == null) {
                log.error("the com not exists, device id : {}", keySyncDTO.getDeviceId());
                return;
            }
            terminalLogger.info("receive the update association key's request : {}", JSONUtil.toJsonStr(keySyncDTO));
            String netAddress = terminalChannelService.getNetAddressByTerminal(com);
            terminalLogger.debug("getRunningDeviceApplicationAssociationState : netAddress {}, deviceAddress {}", netAddress, deviceAddress);
            if (StringUtils.isBlank(netAddress)) {
                return;
            }
            AssociationState associationState = associationManager.getRunningDeviceApplicationAssociationState(netAddress, deviceAddress);
            if (associationState != null) {
                if ("guek".equals(keySyncDTO.getKeyType())) {
                    terminalLogger.info("update guek {}", keySyncDTO.getKeyValue());
                    ((GXDLMSSecureClient) associationState.getClient()).getCiphering().setBlockCipherKey(ByteUtils.hexStringToByte(keySyncDTO.getKeyValue()));
                }
                if ("gak".equals(keySyncDTO.getKeyType())) {
                    terminalLogger.info("update gak {}", keySyncDTO.getKeyValue());
                    ((GXDLMSSecureClient) associationState.getClient()).getCiphering().setAuthenticationKey(ByteUtils.hexStringToByte(keySyncDTO.getKeyValue()));
                }
            }
        }
        acknowledgment.acknowledge();
    }

    @KafkaListener(id = "taskListener", idIsGroup = false, topics = TaskKafkaTopic.TASK_SEND_TO_CON, autoStartup = "false", containerFactory = "batchFactory")
    public void listen(List<ConsumerRecord> dataList, Acknowledgment acknowledgment) {
        for (ConsumerRecord consumerRecord : dataList) {
            taskAllInCounter.increment();
            threadPoolExecutor.execute(() -> {
                doConsume(consumerRecord);
            });
        }
        acknowledgment.acknowledge();
    }

    @Autowired
    TCPClientManager tcpClientManager;

    private void doConsume(ConsumerRecord consumerRecord) {

        ConnectorTask connectorTask = null;
        ComEntity terminal = null;
        try {
            byte[] data = (byte[]) consumerRecord.value();
            connectorTask = ProtoBufferUtil.fromProto(ConnectorTaskProto.ConnectorTask.parseFrom(data), ConnectorTask.class);
//                terminal = terminalChannelService.getTerminalById(connectorTask.getTerminalId());
//                if (terminal == null || terminal.getComId() == null) {
//                    log.debug("terminal not find");
//                    continue;
//                }
            Channel channelByComId = NettyConfigHelper.getChannelByComId(connectorTask.getTerminalId());
            if (channelByComId != null) {
                terminal = channelByComId.attr(NettyConfigHelper.comEntityAttrKey).get();
            }
            if (terminal == null && tcpClientManager.isConnectedInOtherNode(connectorTask.getTerminalId())) {
                log.debug("terminal not in this node");
                return;
            }
            if (terminal == null) {
                terminal = terminalChannelService.getTerminalById(connectorTask.getTerminalId());
                if (terminal == null || ComEntity.CommunicationMode.SERVER.getCode() == Optional.ofNullable(terminal.getComMode()).orElse(-1)) {
                    log.debug("terminal not in this node");
                    return;
                }
            }
        } catch (Exception e) {
            log.debug("getTaskOrTerminalFailed", e);
            return;
        }

        if (connectorTask != null && terminal != null) {
            //只有随抄任务超时
            if ((connectorTask.getFlag() & 8) == 0 || SystemClock.now() - connectorTask.getSendTime() < 30 * DateUtils.MILLIS_PER_MINUTE) {
                Integer ticket = taskTicketManager.getTicketOnce(connectorTask);
                log.debug("getTicketOnce:{}", connectorTask.getTaskNo(), ticket);
                ConnectorTask finalConnectorTask = connectorTask;
                ComEntity finalTerminal = terminal;
                if (ticket == null) {
                    try {
                        //加入内部队列
                        taskTicketManager.addTaskToQueue(finalConnectorTask);
                    } catch (Exception e) {
                        deviceConnectService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, finalConnectorTask.getAtomicTaskNo(), null, "CON task Queue full", finalConnectorTask.getTerminalId(), finalConnectorTask.getFlag(), null), finalTerminal);
                    }
                } else {
                    ConnectorTaskThread connectorTaskThread = new ConnectorTaskThread(finalConnectorTask, finalTerminal, ticket);
                    SpringUtil.getConfigurableBeanFactory().autowireBean(connectorTaskThread);
//                    threadPoolExecutor.execute(connectorTaskThread);
                    connectorTaskThread.run();
                }
            } else {
                log.error("ODRTaskConsumeOverTime,comId:{},taskNo:{},sendTime:{}", connectorTask.getTerminalId(), connectorTask.getTaskNo(), DateFormatUtils.format(new Date(connectorTask.getSendTime()), DateFormatUtils.ISO_8601_EXTENDED_DATE_FORMAT.getPattern()));
            }
        } else {
            log.error("connectorTaskAcceptFailed");
        }
    }
}
